package com.wei.czz.framework.common.helper;

import com.alibaba.fastjson2.JSON;
import com.wei.czz.common.constant.Constant;
import com.wei.czz.common.enums.CommonEnum;
import com.wei.czz.common.enums.common.messageFlow.MessageStatusEnum;
import com.wei.czz.common.enums.rabbitMQ.MqEnum;
import com.wei.czz.common.utils.SecurityUtils;
import com.wei.czz.framework.common.entity.MessageFlowEntity;
import com.wei.czz.framework.common.service.AesService;
import com.wei.czz.framework.common.service.AsyncService;
import com.wei.czz.framework.common.service.MessageFlowService;
import com.wei.czz.framework.common.service.SnowflakeService;
import com.wei.czz.framework.rabbitMQ.message.BaseMessage;
import lombok.AllArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.Optional;

/**
 * Created by IntelliJ IDEA.
 *
 * @author wyw
 * date: 2024-05-12 16:51:49
 * className: RabbitHelper RabbitMQ帮助类
 * version: 1.0
 * description:
 */
@Component
@AllArgsConstructor
public class RabbitHelper {

    private static final Logger log = LoggerFactory.getLogger(RabbitHelper.class);

    private final MessageFlowService messageFlowService;

    private final AesService aesService;

    private final AsyncService asyncService;

    private final RabbitTemplate rabbitTemplate;

    private final SnowflakeService snowflakeService;

    @PostConstruct
    public void init() {
        log.info("开始初始化系统中间件RabbitMQ业务");

        /*
            配置发送消息的回调方法，不管发送成功还是发送失败，都会回调
                回调方法参数说明：
                    1、correlationData：存储回调消息的id及消息相关信息
                    2、ack：            消息发送结果[true-成功 false-失败]
                    3、cause：          消息发送失败原因（发送成功，该值为空）
                几种回调情况：
                    1、发送消息，交换机收到了，回调（ack：true）
                    2、发送消息，交换机接收失败，回调（ack：false，cause：失败原因）
         */
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            log.info("mq消息发送回调参数：correlationData={} ack={} cause={}", correlationData, ack, cause);
            if (!ack && Objects.nonNull(correlationData)) {
                String id = correlationData.getId();
                Long messageId = Long.parseLong(id);
                try {
                    /*
                        修改消息发送记录
                     */
                    MessageFlowEntity messageFlow = new MessageFlowEntity();
                    messageFlow.setId(messageId)
                            .setReplyText(Optional.ofNullable(cause).orElse(StringUtils.EMPTY))
                            .setStatus(MessageStatusEnum.SEND_FAIL.getValue())
                            .setUpdateTime(new Date())
                            .setUpdateUser(StringUtils.EMPTY)
                            .setUpdateUserId(CommonEnum.ZERO.getLongValue());
                    messageFlowService.saveOrEdit(messageFlow);
                } catch (Exception e) {
                    log.error("消息发送失败，更新消息发送记录异常。id={} message={}", id, e.getMessage(), e);
                }
            }
        });

        /*
            当消息由交换机发送到队列时，可能转发失败，则回调该函数，将消息返还给生产者
                可能导致失败的原因：
                    1、队列宕机
                    2、路由键无法对应的队列
                回调方法参数说明：
                    1、returned：封装转发失败返回结果的对象，可以获取相关参数（状态码、提示消息、交换机名称、路由键、发送的消息）
         */
        rabbitTemplate.setReturnsCallback(returned -> {
            try {
                String id = returned.getMessage().getMessageProperties().getHeaders().get(Constant.MESSAGE_ID_KEY).toString();
                log.error("消息转发失败，由交换机返还给生产者。messageId={} replyText={} replyCode={}", id,
                        returned.getReplyText(), returned.getReplyCode());

                Long messageId = Long.parseLong(id);

                /*
                    修改消息发送记录
                 */
                MessageFlowEntity messageFlow = new MessageFlowEntity();
                messageFlow.setId(messageId)
                        .setReplyText(returned.getReplyText())
                        .setStatus(MessageStatusEnum.SEND_FAIL.getValue())
                        .setUpdateTime(new Date())
                        .setUpdateUser(StringUtils.EMPTY)
                        .setUpdateUserId(CommonEnum.ZERO.getLongValue());
                messageFlowService.saveOrEdit(messageFlow);

            } catch (Exception e) {
                log.error("mq转发消息失败回调，更新消息发送记录异常。returned={} message={}", returned, e.getMessage(), e);
            }
        });

        log.info("初始化系统中间件RabbitMQ业务结束");
    }

    /**
     * 发送mq消息
     * @param mqEnum mq枚举对象
     * @param data   消息数据
     */
    public void send(MqEnum mqEnum, Object data) {
        // 获取一个消息记录主键
        Long id = snowflakeService.get();
        // 保存消息发送记录
        asyncService.execute(() -> {
            log.info("异步消息发送，保存发送记录：id={} exchange={} queue={} routingKey={}", id, mqEnum.getExchange(),
                    mqEnum.getQueue(), mqEnum.getRoutingKey());
            // 生成消息发送记录
            MessageFlowEntity messageFlow = this.buildMessageFlow(id, mqEnum, data);
            // 插入记录
            messageFlowService.insert(messageFlow);
        });

        CorrelationData correlationData = new CorrelationData(id.toString());
        // 消息发送
        rabbitTemplate.convertAndSend(mqEnum.getExchange(), mqEnum.getRoutingKey(), data, correlationData);
    }

    /**
     * 发送mq消息
     * @param mqEnum mq枚举对象
     * @param data   消息数据
     */
    public <T extends BaseMessage> void send(MqEnum mqEnum, T data) {
        // 获取一个消息记录主键
        Long id = snowflakeService.get();
        // 保存消息发送记录
        asyncService.execute(() -> {
            log.info("异步消息发送，保存发送记录：id={} exchange={} queue={} routingKey={}", id, mqEnum.getExchange(),
                    mqEnum.getQueue(), mqEnum.getRoutingKey());
            // 生成消息发送记录
            MessageFlowEntity messageFlow = this.buildMessageFlow(id, mqEnum, data);
            // 插入记录
            messageFlowService.insert(messageFlow);
        });
        // 链路追踪值
        String traceId = MDC.get(Constant.TRACE_ID);
        data.setTraceId(traceId);

        CorrelationData correlationData = new CorrelationData(id.toString());
        // 消息发送
        rabbitTemplate.convertAndSend(mqEnum.getExchange(), mqEnum.getRoutingKey(), data, correlationData);
    }

    /**
     * 发送mq消息
     * @param exchange   交换机
     * @param routingKey 路由键
     * @param data       消息数据
     */
    public void send(String exchange, String routingKey, Object data, Long messageId) {
        List<MqEnum> exchangeList = MqEnum.findExchangeList(exchange);
        if (exchangeList.isEmpty()) {
            log.error("交换机未配置枚举。exchange={}", exchange);
            return;
        }
        if (exchangeList.stream().noneMatch(mqEnum -> mqEnum.getRoutingKey().equals(routingKey))) {
            log.error("交换机关联路由键未配置枚举。exchange={} routingKey={}", exchange, routingKey);
            return;
        }
        // 消息发送
        rabbitTemplate.convertAndSend(exchange, routingKey, data, new CorrelationData(messageId.toString()));
    }

    public void sendDelay(MqEnum mqEnum, Object data, int ttl) {
        // 获取一个消息记录主键
        Long id = snowflakeService.get();

        // 保存消息发送记录
        asyncService.execute(() -> {
            log.info("异步延迟消息发送，保存发送记录：id={} exchange={} queue={} routingKey={} ttl={}", id,
                    mqEnum.getExchange(), mqEnum.getQueue(), mqEnum.getRoutingKey(), ttl);
            // 生成消息发送记录
            MessageFlowEntity messageFlow = this.buildMessageFlow(id, mqEnum, data);
            // 设置延迟时长
            messageFlow.setMessageTtl(ttl);
            messageFlowService.insert(messageFlow);
        });

        MessagePostProcessor messagePostProcessor = msg -> {
            MessageProperties messageProperties = msg.getMessageProperties();
            // 手动设置消息的过期时间
            messageProperties.setDelay(ttl);
            return msg;
        };

        CorrelationData correlationData = new CorrelationData(id.toString());
        // 消息发送
        rabbitTemplate.convertAndSend(mqEnum.getExchange(), mqEnum.getRoutingKey(), data, messagePostProcessor,
                correlationData);

    }

    /**
     * 构造消息发送记录
     * @param id     主键
     * @param mqEnum mq枚举对象
     * @param data   消息数据
     * @return 消息发送记录对象
     */
    private MessageFlowEntity buildMessageFlow(Long id, MqEnum mqEnum, Object data) {
        String username = SecurityUtils.getOrDefaultUsername(StringUtils.EMPTY);
        Long userId = SecurityUtils.getOrDefaultUserId(CommonEnum.ZERO.getLongValue());
        Date date = new Date();

        MessageFlowEntity messageFlow = new MessageFlowEntity();
        messageFlow.setId(id)
                .setExchange(mqEnum.getExchange())
                .setQueue(mqEnum.getQueue())
                .setRoutingKey(mqEnum.getRoutingKey())
                .setMessage(aesService.czzEncrypt(JSON.toJSONString(data)))
                .setMessageTtl(CommonEnum.MINUS_ONE.getValue())
                .setReplyText(StringUtils.EMPTY)
                .setStatus(MessageStatusEnum.SUCCESS.getValue())
                .setDeleteStatus(CommonEnum.ZERO.getValue())
                .setUpdateTime(date)
                .setUpdateUser(StringUtils.EMPTY)
                .setUpdateUserId(CommonEnum.ZERO.getLongValue())
                .setCreateTime(date)
                .setCreateUser(username)
                .setCreateUserId(userId);
        return messageFlow;
    }

}
